package cn.doitedu.rtmk.demo3;

import cn.doitedu.rtmk.beans.RuleMetaBean;
import cn.doitedu.rtmk.demo2.UserEvent;
import cn.doitedu.rtmk.utils.RuleHitResultBuilder;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class RuleEngineCoreFunction3 extends KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject> {

    Map<Long, RoaringBitmap> ruleMetasMap = new HashMap<>();


    /**
     * 处理主流数据的方法
     * 静态圈选条件：月平均消费额>200的用户
     * 触发条件：当他发生A行为时（event_id='page_load', properties['url'] = '/page/001'），命中规则
     */
    @Override
    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        if(ruleMetasMap.isEmpty()) return;

        // 取出规则1的预圈选人群
        RoaringBitmap rule1PreSelectUsersBitmap = ruleMetasMap.get(1L);

        // 先判断该用户是否是本规则的目标人群
        if (rule1PreSelectUsersBitmap.contains((int) userEvent.getUser_id())) {
            // 判断该行为是否是本规则的触发条件
            if (userEvent.getEvent_id().equals("page_load") && userEvent.getProperties().get("url").equals("/page/001")) {
                // 如果是，则推送消息
                JSONObject resultObj = new RuleHitResultBuilder()
                        .setUser(userEvent.getUser_id())
                        .setHitTime(userEvent.getEvent_time())
                        .setHitRule(1)
                        .build();

                out.collect(resultObj);
            }
        }

    }

    /**
     * 处理广播流数据的方法
     */
    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 取出新收到的规则元信息中的： 规则id 和 预圈选人群bytes
        long rule_id = ruleMetaBean.getRule_id();
        byte[] pre_select_users = ruleMetaBean.getPre_select_users();

        // 反序列化出预圈选人群的bitmap对象
        RoaringBitmap preSelectUsersBitmap = RoaringBitmap.bitmapOf();
        ByteBuffer byteBuffer = ByteBuffer.wrap(pre_select_users);
        preSelectUsersBitmap.deserialize(byteBuffer);

        log.warn("收到一条新规则的元信息, 规则id:{},规则预圈选人群人数:{}",rule_id,preSelectUsersBitmap.getCardinality());

        // 将加工好的规则元信息，放入全局变量：ruleMetasMap
        ruleMetasMap.put(rule_id, preSelectUsersBitmap);

    }

}
